package com.feidee.fdhadoop.logCollect;

import com.alibaba.fastjson.JSON;
import com.feidee.fdcommon.configuration.CustomConfiguration;
import com.feidee.fdcommon.model.StringResponse;
import com.feidee.fdcommon.util.CustomThreadFactory;
import com.feidee.fdcommon.util.HttpUtil;
import com.feidee.fdfalcon.falcon.FalconConfigName;
import com.feidee.fdfalcon.falcon.FalconCounterType;
import com.feidee.fdfalcon.falcon.FalconReportService;
import com.feidee.fdhadoop.constant.Constant;
import com.google.common.util.concurrent.RateLimiter;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.HttpPost;
import org.apache.http.entity.mime.MultipartEntityBuilder;
import org.apache.http.entity.mime.content.ByteArrayBody;
import org.apache.http.entity.mime.content.ContentBody;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.EnumMap;
import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class LogCollectService {

    protected static final Logger logger = LoggerFactory.getLogger(LogCollectService.class);
    private static String logUrl;
    private static RateLimiter limiterPrintLog = RateLimiter.create(0.1);
    private static RateLimiter limiterOverSize = RateLimiter.create(0.1);
    private static ThreadPoolExecutor threadPool = new ThreadPoolExecutor(CustomConfiguration.getInt("poolSize.LogCollectService", 10), CustomConfiguration.getInt("poolSize.LogCollectService", 100),
            0L, TimeUnit.MILLISECONDS,
            new LinkedBlockingQueue<>(),
            new CustomThreadFactory("LogCollectService"));

    static {
        //配置指标类型
        EnumMap<FalconConfigName, String> enumMap = new EnumMap<>(FalconConfigName.class);
        enumMap.put(FalconConfigName.COUNTER_TYPE, FalconCounterType.GAUGE.getValue());
        FalconReportService.metricConfig("LogCollect.threadPool.queueSize", enumMap);
        logUrl = LogCollectUtils.getLogUrl();
        logger.info("LogCollectService url:{}", logUrl);
    }

    static class LogTask extends Thread {
        private final String content;

        public LogTask(String content) {
            this.content = content;
        }

        @Override
        public void run() {
            FalconReportService.metricIncr("LogCollect.threadPool.run");
            boolean printLog = false;
            if (limiterPrintLog.tryAcquire()) {
                printLog = true;
            }
            HttpPost httpPost = null;
            try {
                if (printLog) {
                    logger.info("LogCollectService send content:{}", content);
                }
                httpPost = new HttpPost(logUrl);
                byte[] contentBytes = content.getBytes("utf-8");
                ContentBody cb = new ByteArrayBody(contentBytes, "");
                HttpEntity entity = MultipartEntityBuilder.create().addPart("content", cb).build();
                httpPost.setEntity(entity);
                StringResponse response = null;
                //发送失败时重试2次
                int retryCount = 0;
                for (int i = 0; i < 3; i++) {
                    try {
                        response = HttpUtil.post(httpPost);
                        break;
                    } catch (Exception e) {
                        if (i == 2) throw e;
                        Thread.sleep(200);
                    }
                    retryCount++;
                }
                if (printLog) {
                    logger.info("LogCollectService response code:{}", response.getStatusCode());
                }
                if (retryCount > 0) {
                    logger.info("LogTask重试{}次后执行成功~", retryCount);
                }
            } catch (Exception e) {
                logger.warn("LogTask重试2次后执行失败,content:" + content, e);
            } finally {
                if (httpPost != null)
                    httpPost.releaseConnection();
            }
        }
    }

    /**
     * @param departmentID
     * @param businessID
     * @param logObj
     * @throws Exception
     */
    public static void sendLog(String departmentID, String businessID, Object logObj) {
        try {
            long queueSize = (long) threadPool.getQueue().size();
            FalconReportService.metricSet("LogCollect.threadPool.queueSize", queueSize);
            if (queueSize > Constant.LOGCOLLECT_THREADPOOL_MAX_QUEUESIZE) {
                if (limiterOverSize.tryAcquire()) {
                    logger.info("LogCollectService queueSize达到{},直接返回~", Constant.LOGCOLLECT_THREADPOOL_MAX_QUEUESIZE);
                }
                return;
            }
            HashMap<String, String> commons = new HashMap<>();
            commons.put("departmentID", departmentID);
            commons.put("businessID", businessID);

            HashMap<String, Object> logMap = new HashMap<>();
            logMap.put("commons", commons);

            if (List.class.isInstance(logObj)) {
                logMap.put("events", logObj);
            } else {
                LinkedList<Object> events = new LinkedList<Object>();
                events.add(logObj);
                logMap.put("events", events);
            }

            String content = JSON.toJSONString(logMap);
            threadPool.execute(new LogTask(content));
            FalconReportService.metricIncr("LogCollect.threadPool.submit");
        } catch (Exception e) {
            logger.error("LogCollectService sendLog 异常~", e);
        }
    }
}
